

package main

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"time"

	"go.etcd.io/etcd/client/pkg/v3/fileutil"
	"go.etcd.io/etcd/client/pkg/v3/types"
	"go.etcd.io/etcd/raft/v3"
	"go.etcd.io/etcd/raft/v3/raftpb"
	"go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
	"go.etcd.io/etcd/server/v3/etcdserver/api/snap"
	stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
	"go.etcd.io/etcd/server/v3/storage/wal"
	"go.etcd.io/etcd/server/v3/storage/wal/walpb"

	"go.uber.org/zap"
)

type commit struct {
	data       []string
	applyDoneC chan<- struct{}
}

// raft支持的关键值流
type raftNode struct {
	proposeC    <-chan string            // 建议的消息（k，v）
	confChangeC <-chan raftpb.ConfChange // 建议的群集配置更改
	commitC     chan<- *commit           // 提交到日志的条目（k，v）
	errorC      chan<- error             // 来自raft会话的错误

	id          int      // raft会话的客户端ID 
	peers       []string // raft对等URL 
	join        bool     // 节点正在加入现有群集
	waldir      string   // WAL目录路径
	snapdir     string   // 快照目录路径
	getSnapshot func() ([]byte, error)

	confState     raftpb.ConfState
	snapshotIndex uint64
	appliedIndex  uint64

	// 提交/错误通道
	node        raft.Node
	raftStorage *raft.MemoryStorage
	wal         *wal.WAL

	snapshotter      *snap.Snapshotter
	snapshotterReady chan *snap.Snapshotter // 当Snapshoter准备就绪时发出信号

	snapCount uint64
	transport *rafthttp.Transport
	stopc     chan struct{} // 发出建议通道关闭
	httpstopc chan struct{} // 发出信号http服务器关闭
	httpdonec chan struct{} // 发出信号http服务器关闭完成

	logger *zap.Logger
}

var defaultSnapshotCount uint64 = 10000

// 在该方法中会使用上层模块传入的配置信息（其中包括proposeC通道和confChangeC通道）来创建ra企Node实例，
// 同时会创建commitC通道和errorC通道返回给上层模块使用。这样，上层模块就可以通过这几个通道与rafeNode实例进行交互了。
// 另外，newRaftNode（）函数中还会启动一个独立的后台goroutine来完成回放WAL日志、启动网络组件等初始化操作。
func newRaftNode(id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
	confChangeC <-chan raftpb.ConfChange) (<-chan *commit, <-chan error, <-chan *snap.Snapshotter) {
	// 创建commitC和errorC通道
	commitC := make(chan *commit)
	errorC := make(chan error)

	rc := &raftNode{
		proposeC:    proposeC,
		confChangeC: confChangeC,
		commitC:     commitC,
		errorC:      errorC,
		id:          id,
		peers:       peers,
		join:        join,
		// 初始化存放WAL日志和SnapShaot文件的目录
		waldir:      fmt.Sprintf("raftexample-%d", id),
		snapdir:     fmt.Sprintf("raftexample-%d-snap", id),
		getSnapshot: getSnapshot,
		snapCount:   defaultSnapshotCount,	// 默认是10000
		// 创建 stopc,httostopc,httpdonec和 snapshotterReady四个通道
		stopc:       make(chan struct{}),
		httpstopc:   make(chan struct{}),
		httpdonec:   make(chan struct{}),

		logger: zap.NewExample(),

		snapshotterReady: make(chan *snap.Snapshotter, 1),
		// 其余字段在WAL日志回放完成之后才会初始化
		// WAL重播后填充的其余结构
	}
	// 单独启动一个goroutine执行startRaft()方法，在该方法中完成剩余初始化操作
	go rc.startRaft()
	// 将 commitC,errorC,snapshotterReady三个通道返回给上层应用
	return commitC, errorC, rc.snapshotterReady
}

// 将新生成的快照数据保存到磁盘上，还会根据快照的元数据释放部分旧WAL日志文件的句柄
func (rc *raftNode) saveSnap(snap raftpb.Snapshot) error {
	// 根据快照的元数据，创建 walpb.Snapshot实例
	walSnap := walpb.Snapshot{
		Index:     snap.Metadata.Index,
		Term:      snap.Metadata.Term,
		ConfState: &snap.Metadata.ConfState,
	}
	// 在将快照写入WAL之前保存快照文件。
	// 这使快照文件成为孤立文件成为可能，但可防止WAL快照条目没有相应的快照文件。
	if err := rc.snapshotter.SaveSnap(snap); err != nil {
		return err
	}
	// WAL会将上述快照的元数据信息封装成一条日志记录下来
	if err := rc.wal.SaveSnapshot(walSnap); err != nil {
		return err
	}
	// 根据快照的元数据信息，释放一些无用的WAL日志文件的句柄，
	return rc.wal.ReleaseLockTo(snap.Metadata.Index)
}

// 对Ready实例中携带的待应用记录进行过滤
func (rc *raftNode) entriesToApply(ents []raftpb.Entry) (nents []raftpb.Entry) {
	if len(ents) == 0 {
		return ents
	}
	// 检测ents长度
	firstIdx := ents[0].Index
	// 检测firstIndex是否合法
	if firstIdx > rc.appliedIndex+1 {
		log.Fatalf("first index of committed entry[%d] should <= progress.appliedIndex[%d]+1", firstIdx, rc.appliedIndex)
	}
	// 过滤掉已经被应用过的Entry记录
	if rc.appliedIndex-firstIdx+1 < uint64(len(ents)) {
		nents = ents[rc.appliedIndex-firstIdx+1:]
	}
	return nents
}

// raftNode会将所有待应用记录写入cornrnitC通道中。
// 后续kvstore就可以读取cornrnitC通道并保存相应的键值对数据
func (rc *raftNode) publishEntries(ents []raftpb.Entry) (<-chan struct{}, bool) {
	if len(ents) == 0 {
		return nil, true
	}

	data := make([]string, 0, len(ents))
	for i := range ents {
		switch ents[i].Type {
		case raftpb.EntryNormal:
			// 如果Entry记录的Data为空，则直接忽略该条Entry记录
			if len(ents[i].Data) == 0 {
				// 忽略空消息
				break
			}
			s := string(ents[i].Data)
			data = append(data, s)
		case raftpb.EntryConfChange:
			var cc raftpb.ConfChange	// 将EntryConfChange类型的记录封装成ConfChange
			cc.Unmarshal(ents[i].Data)
			// 将ConfChange实例传入底层的etcd-raft组件
			rc.confState = *rc.node.ApplyConfChange(cc)
			// 除了etcd-raft纽件中需要创建（或删除）对应的Progress实例，网络层也需要做出相应的调
			// 整，即添加（或删除）相应的Peer实例
			switch cc.Type {
			case raftpb.ConfChangeAddNode:
				if len(cc.Context) > 0 {
					rc.transport.AddPeer(types.ID(cc.NodeID), []string{string(cc.Context)})
				}
			case raftpb.ConfChangeRemoveNode:
				if cc.NodeID == uint64(rc.id) {
					log.Println("I've been removed from the cluster! Shutting down.")
					return nil, false
				}
				rc.transport.RemovePeer(types.ID(cc.NodeID))
			}
		}
	}

	var applyDoneC chan struct{}

	if len(data) > 0 {
		applyDoneC = make(chan struct{}, 1)
		select {
		// 将数据写入commitC通道，kvstore会读取从其中读取并记录相应的KV值
		case rc.commitC <- &commit{data, applyDoneC}:
		case <-rc.stopc:
			return nil, false
		}
	}

	// 提交后，更新appliedIndex 
	rc.appliedIndex = ents[len(ents)-1].Index

	return applyDoneC, true
}

func (rc *raftNode) loadSnapshot() *raftpb.Snapshot {
	if wal.Exist(rc.waldir) {
		walSnaps, err := wal.ValidSnapshotEntries(rc.logger, rc.waldir)
		if err != nil {
			log.Fatalf("raftexample: error listing snapshots (%v)", err)
		}
		snapshot, err := rc.snapshotter.LoadNewestAvailable(walSnaps)
		if err != nil && err != snap.ErrNoSnapshot {
			log.Fatalf("raftexample: error loading snapshot (%v)", err)
		}
		return snapshot
	}
	return &raftpb.Snapshot{}
}

// raftNode.openWAL（）方法中确定读取WAL日志文件的位置及创建WAL实例
func (rc *raftNode) openWAL(snapshot *raftpb.Snapshot) *wal.WAL {
	if !wal.Exist(rc.waldir) {	// 检测 WAL 日志目录是否存在,如果不存在则进行创建
		if err := os.Mkdir(rc.waldir, 0750); err != nil {
			log.Fatalf("raftexample: cannot create dir for wal (%v)", err)
		}

		// 新建 WAL 实例,其中会船舰相应的目录和一个空的 WAL 日志文件
		w, err := wal.Create(zap.NewExample(), rc.waldir, nil)
		if err != nil {
			log.Fatalf("raftexample: create wal error (%v)", err)
		}
		w.Close()	// 关闭 WAL ,其中包括跟中关闭目录,文件和相关的 goroutine
	}

	// 创建 walsnap.Snapshot 实例并初始化其 Index 字段和 Term 字段,注意两者的区别:
	// walsnap.Snapshot 只包含了快照元数据中的 Term 值和索引值,并不会包含真正的快照数据
	walsnap := walpb.Snapshot{}
	if snapshot != nil {
		walsnap.Index, walsnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
	}
	log.Printf("loading WAL at term %d and index %d", walsnap.Term, walsnap.Index)
	w, err := wal.Open(zap.NewExample(), rc.waldir, walsnap)	// 创建 WAL 实例
	if err != nil {
		log.Fatalf("raftexample: error loading wal (%v)", err)
	}

	return w
}

// raftNode.replayWAL（）方法首先会读取快照数据，在快照的元数据中记录了该快照包含的最后一条Entry记录的Term值和索引值。
// 然后根据该Term值和索引值确定读取WAL日志文件的位置，并进行日志记录的读取。
// 最后将读取到的快照数据、WAL日志记录和状态信息保存到raftNode.raftStorage（即巳tcd-raft模块中raftLog.storage）中
func (rc *raftNode) replayWAL() *wal.WAL {
	log.Printf("replaying WAL of member %d", rc.id)
	// 读取快照文件,该方法底层会调用 Snapshotter.Load() 方法完成快照文件的读取,
	snapshot := rc.loadSnapshot()
	w := rc.openWAL(snapshot)	// 根据读取到的 SnapShot 实例的元数据创建 WAL 实例
	_, st, ents, err := w.ReadAll()	// 读取快照数据之后的全部 WAL 日志数据,并获取状态信息
	// 异常检查,若读取 WAL 日志文件的过程中出现异常,则输出日志并终止程序
	if err != nil {
		log.Fatalf("raftexample: failed to read WAL (%v)", err)
	}
	// 创建 MemoryStorage 实例,
	rc.raftStorage = raft.NewMemoryStorage()
	if snapshot != nil {
		rc.raftStorage.ApplySnapshot(*snapshot)	// 将快照数据加载到 MemoryStorage 中
	}
	// 将读取 WAL 日志之后得到的 HardState 加载到 MemoryStorage 中
	rc.raftStorage.SetHardState(st)

	// 将读取 WAL 日志得到的 Entry 记录加载到 MemoryStorage 中
	// 附加到存储器，使 raft 在日志中的正确位置启动
	rc.raftStorage.Append(ents)

	return w
}

func (rc *raftNode) writeError(err error) {
	rc.stopHTTP()
	close(rc.commitC)
	rc.errorC <- err
	close(rc.errorC)
	rc.node.Stop()
}

// startRaft（）函数，其核心步骤如下：
// (1）创建Snapshotter，并将该Snapshotter实例返回给上层模块。
// (2）创建WAL实例，然后加载快照井回放WAL日志。
// (3）创建raft.Config实例，其中包含了启动etcd-raft模块的所有配置，
// (4）初始化底层etcd-raft模块，得到前面介绍的node实例。
// (5）创建Transport实例，该实例负责集群中各个节点之间的网络通信，其具体实现在rafthttp包中。
// (6）建立与集群中其他节点的网络连接。
// (7）启动网络组件，其中会监昕当前节点与集群中其他节点之间的网络连接，并进行节点之间的消息读写。
// (8）启动两个后台goroutine，它们的主要工作都是处理上层模块与底层etcd-raft模块的交互，但处理的具体内容不同，
func (rc *raftNode) startRaft() {
	// 检测 snapdir 字段指定的目录是否存在,该目录用于存放定期生成的快照数据
	// 若 snapdir 目录不存在,则进行创建,若创建失败,则输出异常日志并终止程序
	if !fileutil.Exist(rc.snapdir) {
		if err := os.Mkdir(rc.snapdir, 0750); err != nil {
			log.Fatalf("raftexample: cannot create dir for snapshot (%v)", err)
		}
	}
	// 步骤1,创建 Snapshotter 实例,并该 Snapshotter 实例通过 snapshotterReady 通道返回给上层
	// 应用 SnapShotter 实例体统了读写快照文件的功能,
	rc.snapshotter = snap.New(zap.NewExample(), rc.snapdir)

	// 步骤2,创建 WAL 实例,然后加载快照并回访 WAL 日志
	oldwal := wal.Exist(rc.waldir)	// 检测 waldir 目录下是否存在旧的 WAL 日志文件
	rc.wal = rc.replayWAL()	// 在 replayWAL() 方法中会先加载快照数据,然后重放 WAL 日志文件

	// 信号重放已完成
	rc.snapshotterReady <- rc.snapshotter

	// 步骤3,创建 raft.Config 实例
	rpeers := make([]raft.Peer, len(rc.peers))
	for i := range rpeers {
		rpeers[i] = raft.Peer{ID: uint64(i + 1)}
	}
	c := &raft.Config{
		ID:                        uint64(rc.id),
		ElectionTick:              10,	// election timeout 时间
		HeartbeatTick:             1,	// 心跳计时器的超时时间
		// 持久化存储,与 etcd-raft 模块中的 raftLog.storage 共享同一个 MemoryStorage 实例
		Storage:                   rc.raftStorage,
		MaxSizePerMsg:             1024 * 1024,	// 每条消息的最大长度
		MaxInflightMsgs:           256,	// 已发但是收到未响应的消息上限个数
		MaxUncommittedEntriesSize: 1 << 30,
	}

	// 步骤4,初始化底层 etcd-raft 模块,这里会根据 WAL 日志的回放情况,判断当前几点是首次启动
	// 还是重新启动, StartNode() 和 RestartNode() 函数略
	if oldwal || rc.join {
		rc.node = raft.RestartNode(c)
	} else {
		rc.node = raft.StartNode(c, rpeers)
	}

	// 步骤5,创建 Transport 实例并启动,他负责 raft 节点之间通信的网络服务,
	rc.transport = &rafthttp.Transport{
		Logger:      rc.logger,
		ID:          types.ID(rc.id),
		ClusterID:   0x1000,
		Raft:        rc,
		ServerStats: stats.NewServerStats("", ""),
		LeaderStats: stats.NewLeaderStats(zap.NewExample(), strconv.Itoa(rc.id)),
		ErrorC:      make(chan error),
	}
	rc.transport.Start()	// 启动网络服务相关的组件

	// 步骤6,建立与集群中的其他各个节点的连接,
	for i := range rc.peers {
		if i+1 != rc.id {
			rc.transport.AddPeer(types.ID(i+1), []string{rc.peers[i]})
		}
	}

	// 步骤7,启动一个 goroutine ,其中会监听当前节点与集群中其他节点之间的网络连接,
	go rc.serveRaft()
	// 步骤8,启动后台 goroutine 处理上层应用与底层 etcd-raft 模块的交互
	go rc.serveChannels()
}

// 停止关闭http，关闭所有通道，并停止raft。
func (rc *raftNode) stop() {
	rc.stopHTTP()
	close(rc.commitC)
	close(rc.errorC)
	rc.node.Stop()
}

func (rc *raftNode) stopHTTP() {
	rc.transport.Stop()
	close(rc.httpstopc)
	<-rc.httpdonec
}

// 通知上层模块加载新生成的快照数据，并使用新快照的元数据更新raftNode中的相应字段
func (rc *raftNode) publishSnapshot(snapshotToSave raftpb.Snapshot) {
	if raft.IsEmptySnap(snapshotToSave) {
		return
	}

	log.Printf("publishing snapshot at index %d", rc.snapshotIndex)
	defer log.Printf("finished publishing snapshot at index %d", rc.snapshotIndex)

	if snapshotToSave.Metadata.Index <= rc.appliedIndex {
		log.Fatalf("snapshot index [%d] should > progress.appliedIndex [%d]", snapshotToSave.Metadata.Index, rc.appliedIndex)
	}
	// 使用 commitC 通道通知上层应用加载新生成的快照数据
	rc.commitC <- nil // 触发kvstore加载快照

	// 记录新快照的元数据
	rc.confState = snapshotToSave.Metadata.ConfState
	rc.snapshotIndex = snapshotToSave.Metadata.Index
	rc.appliedIndex = snapshotToSave.Metadata.Index
}

var snapshotCatchUpEntriesN uint64 = 10000

// 释放底层etcd-raft模块中无用的Entry记录，
// 节点每处理指定条数（默认是10000条）的记录之后，就会触发一次快照生成操作
func (rc *raftNode) maybeTriggerSnapshot(applyDoneC <-chan struct{}) {
	// 检测处理的记录数是否足够，如果不足，则直接返回
	if rc.appliedIndex-rc.snapshotIndex <= rc.snapCount {
		return
	}

	// 等待所有提交的条目被应用（或服务器关闭）
	if applyDoneC != nil {
		select {
		case <-applyDoneC:
		case <-rc.stopc:
			return
		}
	}

	log.Printf("start snapshot [applied index: %d | last snapshot index: %d]", rc.appliedIndex, rc.snapshotIndex)
	// 获取快照数据，在raftexarnple示例中是获取kvstore中记录的全部键位对数据，
	data, err := rc.getSnapshot()
	if err != nil {
		log.Panic(err)
	}
	// 创建Snapshot实例，同时也会将快照和元数据更新到raftLog.MernoryStorage中
	snap, err := rc.raftStorage.CreateSnapshot(rc.appliedIndex, &rc.confState, data)
	if err != nil {
		panic(err)
	}
	// 保存快照数据，
	if err := rc.saveSnap(snap); err != nil {
		panic(err)
	}

	// 计算压缩的位置，压缩之后，该位置之前的全部记录都会被抛弃
	compactIndex := uint64(1)
	if rc.appliedIndex > snapshotCatchUpEntriesN {
		compactIndex = rc.appliedIndex - snapshotCatchUpEntriesN
	}
	// 压缩raftLog中保存的Entry记录
	if err := rc.raftStorage.Compact(compactIndex); err != nil {
		panic(err)
	}

	log.Printf("compacted log at index %d", compactIndex)
	// 更新raftNode的相关字段
	rc.snapshotIndex = rc.appliedIndex
}

// serveChannels（）方法，其中会单独启动一个后台goroutine来负责上层模块传递给etcd-raft模块的数据，
// 主要处理前面介绍的proposeC、confChangeC两个通道。
// 还负责处理etcd-ra位模块返回给上层模块的数据及其他相关的操作，
//例如，推进etcd-raft模块的逻辑时钟、处理网络异常等，
func (rc *raftNode) serveChannels() {
	// 获取快照数据和快照元数据
	snap, err := rc.raftStorage.Snapshot()
	if err != nil {
		panic(err)
	}
	rc.confState = snap.Metadata.ConfState
	rc.snapshotIndex = snap.Metadata.Index
	rc.appliedIndex = snap.Metadata.Index

	defer rc.wal.Close()

	// 创建一个每隔 1OOms 触发一次的定时器，那么在逻辑上，1OOms 即是 etcd-raft 纽件的最小时间单位，
	// 该定时器每触发一次，则逻辑时钟推进一次
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()

	// 单独启动一个 goroutine 负责将 proposeC、confChangeC 通道上接收到的数据传递给 etcd-raft
	// 纽件进行处理
	go func() {
		confChangeCount := uint64(0)

		for rc.proposeC != nil && rc.confChangeC != nil {
			select {
			case prop, ok := <-rc.proposeC:	// 收到上层应用通过 proposeC 通道传递过来的数据
				// 如采发生异常，则将 raftNode.proposeC 字段置空，当前循环及整个 goroutine 都会结束
				// 通过 node.Propose() 方法，将数据传入底层 etcd-raft 纽件进行处理
				if !ok {
					rc.proposeC = nil
				} else {
					// 块，直到raft状态机
					rc.node.Propose(context.TODO(), []byte(prop))
				}

			case cc, ok := <-rc.confChangeC:	// 收到上层应用通过 confChangeC 通道传递过来的数据
				// 如果发生异常，则将 raftNode.confChangeC 字段直空
				if !ok {
					rc.confChangeC = nil
				} else {
					confChangeCount++	// 统计集群交史请求的个数，并将其作为 ID
					cc.ID = confChangeCount
					// 通过 node.ProposeConfChange() 方法，将数据传入底层 etcd-raft 纽件进行处理，
					rc.node.ProposeConfChange(context.TODO(), cc)
				}
			}
		}
		// 关闭 stopc 通道，触发 rafeNode.stop() 方法的调用
		close(rc.stopc)
	}()

	// 该循环主要负责处理底层 etcd-raft 纽件返回的 Ready 数据
	for {
		select {
		case <-ticker.C:
			rc.node.Tick()	// 上述 ticker 定时器触发一次，即会推进 etcdraft 纽件的逻辑时钟

		// 读取 node.readyc 迢迢，前面介绍 etcd-raft 纽件时也提到，该通道是 etcdraft 纽件与上层应用
		// 交互的主要通道之一，其中传递的 Ready 实例也封装了很多信
		case rd := <-rc.node.Ready():
			// 将当前 etcdraft 纽件的状态信息，以及待持久化的 Entry 记录先记录到 WAL 日志文件中，
			// 即使之后脱机，这些信息也可以在节点下次启动时，通过前面回放 WAL 日志的方式进行恢复
			rc.wal.Save(rd.HardState, rd.Entries)
			if !raft.IsEmptySnap(rd.Snapshot) {	// 检测 etcd-raft 纽件生成了新的快照数据
				rc.saveSnap(rd.Snapshot)	// 将新的快照数据写入快照文件中
				rc.raftStorage.ApplySnapshot(rd.Snapshot)	// 将新快照持久化到 raftStorage,
				rc.publishSnapshot(rd.Snapshot)	// 通知上层应用加载新快照
			}
			rc.raftStorage.Append(rd.Entries)	// 将待持久化的 Entry 记录追加到 raftStorage 中完成持久化
			rc.transport.Send(rd.Messages)	// 将待发送的消息发送到指定节点，
			// 将已提交、待应用的 Entry 记录应用到上层应用的状态机中
			applyDoneC, ok := rc.publishEntries(rc.entriesToApply(rd.CommittedEntries))
			if !ok {
				rc.stop()
				return
			}
			// 随着节点的运行，WAL 日志量和 raftLog.storage 中的 Entry 记录会不断增加，
			// 所以节点每处理10000条（默认值）Entry 记录，就会触发一次创建快照的过程，
			// 同时 WAL 会释放一些日志文件的句柄，raftLog.storage 也会压缩其保存的 Entry 记录
			rc.maybeTriggerSnapshot(applyDoneC)
			// 上层应用处理完该 Ready 实例，通知 etcd-raft 纽件准备返回下一个 Ready 实例
			rc.node.Advance()

		case err := <-rc.transport.ErrorC:	// 处理网络异常
			rc.writeError(err)	// 关闭与集群中其他节点的网络连接
			return

		case <-rc.stopc:	// 处理关闭命令
			rc.stop()
			return
		}
	}
}

// serveRaft（）方法负责监听当前节点的地址，完成与其他节点的通信。
func (rc *raftNode) serveRaft() {
	url, err := url.Parse(rc.peers[rc.id-1])	// 获取当前系欸但的 URL 地址
	if err != nil {
		log.Fatalf("raftexample: Failed parsing URL (%v)", err)
	}

	// 创建 stoppableListener 实例, stoppableListener 继承了 net.TCPListener(淡然也实现了
	// net.listener 接口),他会与 http.Server 配合实现对当前节点的 URL 地址进行监听
	ln, err := newStoppableListener(url.Host, rc.httpstopc)
	if err != nil {
		log.Fatalf("raftexample: Failed to listen rafthttp (%v)", err)
	}

	// 创建 http.Server 实例，会通过上面的 stoppableListener 实例监听当前界定啊的 URL 地址
	// stoppobleListener.Accept() 方法监听到新连接时,会创建对应的 net.Conn 实例, http.Server
	// 会为每一个连接船舰单独的 goroutine 处理,每个请求都会由 http.Server.Handler 处理,这里的
	// Handler 是由 Rafthttp.Transporter 创建的,
	// http.Server.Serve() 方法会一直阻塞,直到 http.Server 关闭
	err = (&http.Server{Handler: rc.transport.Handler()}).Serve(ln)
	select {
	case <-rc.httpstopc:
	default:
		log.Fatalf("raftexample: Failed to serve rafthttp (%v)", err)
	}
	close(rc.httpdonec)
}

func (rc *raftNode) Process(ctx context.Context, m raftpb.Message) error {
	return rc.node.Step(ctx, m)
}
func (rc *raftNode) IsIDRemoved(id uint64) bool  { return false }
func (rc *raftNode) ReportUnreachable(id uint64) { rc.node.ReportUnreachable(id) }
func (rc *raftNode) ReportSnapshot(id uint64, status raft.SnapshotStatus) {
	rc.node.ReportSnapshot(id, status)
}
